package com.atguigu;

import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.alibaba.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author:GaoFei
 * @Description:
 * @Date:Created in 18:11
 * @Modified By:
 */
public class FlinkCDC {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
//        System.setProperty("HADOOP_USER_NAME","atguigu");
//        env.setStateBackend(new FsStateBackend("hdfs://192.168.10.100:8020/gmall-flink/ck"));
////        System.setProperty("HADOOP_USER_NAME","atguigu");
//        // 设置checkpoint时间
//        env.enableCheckpointing(5000L);
//        // 设置精准一次性
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        // 设置check超时时间
//        env.getCheckpointConfig().setCheckpointTimeout(10000L);
//        // 设置最多两个checkpoint一块
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
//        // 设置两个check间隔
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
        DebeziumSourceFunction<String> build = MySQLSource.<String>builder()
                .hostname("192.168.10.100")
                .port(3306)
                .username("root")
                .password("admin")
                .databaseList("gmall-0625-flink")
                .tableList("gmall-0625-flink.base_trademark")
                .deserializer(new FlinkSerializeDefine())
                .startupOptions(StartupOptions.initial())
                .build();

        DataStreamSource<String> result = env.addSource(build);
        result.print();
        env.execute("FlinkCDC");
    }
}
